package com.it.stream;

import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class IntervalJoinTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = executionEnvironment.fromElements(new Event("tom", "./index", 1000L),
                        new Event("tom", "./product?id=10", 2000L),
                        new Event("lily", "./index", 1000L),
                        new Event("tom", "./product?id=20", 3000L),
                        new Event("tom", "./product?id=30", 4000L),
                        new Event("tom", "./product?id=20", 5000L),
                        new Event("tom", "./order?id=30", 6000L),
                        new Event("tom", "./product?id=50", 7000L),
                        new Event("tom", "./product?id=60", 8000L),
                        new Event("tom", "./product?id=70", 9000L),
                        new Event("lily", "./product?id=100", 7000L))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
        SingleOutputStreamOperator<Tuple2<String, Long>> orderSource = executionEnvironment.fromElements(Tuple2.of("tom", 7000L)).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                return element.f1;
            }
        }));
        orderSource.keyBy(data -> data.f0).intervalJoin(eventSource.keyBy(data -> data.user))
                .between(Time.seconds(-4L), Time.seconds(1L)).process(new ProcessJoinFunction<Tuple2<String, Long>, Event, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> left, Event right, ProcessJoinFunction<Tuple2<String, Long>, Event, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect(new StringBuilder().append("user：").append(left.f0).append("下了一个单，时间为：").append(left.f1).append(", 最近访问url为：").append(right.url).append(" , time is ").append(right.timestamp).toString());

                    }
                }).print("near view urls ");
        executionEnvironment.execute();
    }
}
